package cn.itcast.avro;

import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Map;
// 1.泛型参数继承avro基类，实现序列化接口
public class AvroSerializer<T extends SpecificRecordBase> implements Serializer<T> {
    /**
     *
     *  2.重写序列化方法
     *  3.新建字节数组输出流对象
     *  4.获取二进制对象BinaryEncoder
     *  5.输出数据
     */
    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public byte[] serialize(String s, T t) {
        SpecificDatumWriter<T> userSpecificDatumWriter = new SpecificDatumWriter<>(t.getSchema());
        ByteArrayOutputStream bos=new ByteArrayOutputStream();
        BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(bos, null);
        try {
            userSpecificDatumWriter.write(t,binaryEncoder);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return bos.toByteArray();
    }

    @Override
    public void close() {

    }
}
